/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package overfeatfeatureextraction;

import avroserialization.AvroSerialization;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author mwh
 */
public class HBaseGetter {

    HTable imageTable;
    HTable queueTable;
    HTable failedTable;
    Random random;
    AvroSerialization avroByteSerializer;
    
    public static void main(String[] args) throws Exception {
        HBaseGetter hbaseGetter = new HBaseGetter();
        List<AvroImage> images = new ArrayList<>();
        hbaseGetter.get(8, images);
        System.err.println(images.size());
        
//        System.err.println(images);
    }

    public HBaseGetter() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        imageTable = new HTable(conf, "images");
        queueTable = new HTable(conf, "imagequeue");
        failedTable = new HTable(conf, "imagefailed");
        random = new Random();
        avroByteSerializer = AvroSerialization.INSTANCE;
    }

    public List<byte[]> get(int max, List<AvroImage> resultList) throws IOException {



        Scan scan = new Scan();
        ResultScanner resultScanner = queueTable.getScanner(scan);
        List<byte[]> keys = new ArrayList<>();
        int c = 0;
        for (Result result : resultScanner) {
            byte[] key = result.getRow();
            keys.add(key);
            c++;
            if (c == max) {
                break;
            }
        }

        // Delete from queue table
        deleteFromTable(keys, queueTable);

        // We assume that the job will fail so add to failed table
        // if the job succeeds they will be deleted from failed table
        addToTable(keys, failedTable);

        List<Get> gets = new ArrayList<>();
        for (byte[] key : keys) {
            Get get = new Get(key);
            gets.add(get);
        }
        Result[] results = imageTable.get(gets);

        List<AvroImage> images = new ArrayList<>();
        for (Result result : results) {
            byte[] content = result.getValue(Bytes.toBytes("d"), Bytes.toBytes("avroimage"));
            ByteArrayInputStream bais = new ByteArrayInputStream(content);
            AvroImage avroImage = (AvroImage) avroByteSerializer.desirializeWithSchema(bais, AvroImage.getClassSchema());
            
            
//            AvroImage avroImage = AvroImage.newBuilder().
//                    setMd5(ByteBuffer.wrap(result.getRow())).
//                    setContent(ByteBuffer.wrap(content)).
//                    build();
            images.add(avroImage);
        }
        resultList.addAll(images);
        return keys;

    }

    public void deleteFromTable(List<byte[]> keys, HTable htable) throws IOException {
        List<Delete> deletes = new ArrayList<>();
        for (byte[] key : keys) {
            Delete delete = new Delete(key);
            deletes.add(delete);
        }
        htable.delete(deletes);

    }

    public void addToTable(List<byte[]> keys, HTable htable) throws IOException {
        List<Put> puts = new ArrayList<>();
        for (byte[] key : keys) {
            Put put = new Put(key);

            Date d = new Date(System.currentTimeMillis());
            put.add(Bytes.toBytes("d"), Bytes.toBytes("d"), Bytes.toBytes(d.toString()));
            puts.add(put);
        }
        htable.put(puts);

    }
}
